package com.huoshan.im.server.service;

import com.huoshan.im.server.SessionHolder;
import com.huoshan.im.server.config.properties.SettingConfig;
import com.huoshan.im.server.dto.UserDto;
import com.huoshan.im.server.messaging.MessageId;
import com.huoshan.im.server.utils.FeiShuUtil;
import com.qy.esports.common.utils.JsonUtils;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import online.chazi.netkit.Session;
import online.chazi.netkit.channel.NetkitFuture;
import online.chazi.netkit.channel.NetkitFutureListener;
import online.chazi.netkit.messaging.Message;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.loadbalancer.LoadBalancerClient;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;

/**
 * 说明：作者很懒，什么都没留下
 * Created by 火山九日 on 2020-04-27 13:32
 */
@Slf4j
@Service
@AllArgsConstructor
public class MessageSenderService {

    private FeiShuUtil feiShuUtil;
    private final SettingConfig settingConfig;
    private final LoadBalancerClient loadBalancerClient;
    private final MessageKafkaService messageKafkaService;
    private final RedisTemplate<String, Object> redisTemplate;

    /**
     * 发送消息给用户
     *
     * @param fromUserId 发送者的用户id
     * @param toUserId   接手消息的目标用户id
     * @param content    消息内容
     */
    public void sendToUser(Long fromUserId, long toUserId, String content) {
        final Message message = new Message(MessageId.SRV_SEND_MESSAGE.id, content);
        this.sendToUser(fromUserId, toUserId, null, message);
    }

    /**
     * @param fromUserId
     * @param toUserId
     * @param toClientType
     * @param message
     */
    public void sendToUser(Long fromUserId, long toUserId, String toClientType, Message message) {
        UserDto userDto = new UserDto(toUserId, toClientType);

        Session session = SessionHolder.getUserSession(userDto);

        if (session != null) {
            //接收人在本server
            NetkitFuture future = session.sendMessage(message);
            future.addFutureListener(new NetkitFutureListener() {
                @Override
                public void onComplete(boolean success, Session session) {
                    if (!success) {
                        //发送失败
                        messageKafkaService.unSendMessage(fromUserId, toUserId, toClientType, message);
                    }
                }
            });
        } else {
            //不在本server
            messageKafkaService.send(fromUserId, toUserId, toClientType, message);
        }
    }


    /**
     * @param toUserId
     * @param message
     */
    public void sendToUserUnKafka(Long fromUserId, long toUserId, String toClientType, Message message) {
        Session session = SessionHolder.getUserSession(toUserId, toClientType);
        if (session != null) {
            //接收人在本server
            NetkitFuture future = session.sendMessage(message);
            future.addFutureListener(new NetkitFutureListener() {
                @Override
                public void onComplete(boolean success, Session session) {
                    if (!success) {
                        //发送失败
                        messageKafkaService.unSendMessage(fromUserId, toUserId, toClientType, message);
                    }
                }
            });
        } else {
            //不在本server
            log.error("不在本server,toUserId={},toClientType={}", toUserId, toClientType);
        }
    }

    /**
     * 发送消息给repeater
     *
     * @param message 需要转发的消息
     */
    public void sendToRepeater(Message message) {
        sendToRepeater(message, 0);
    }

    /**
     * 发送消息给repeater
     *
     * @param message 需要转发的消息
     */
    private void sendToRepeater(Message message, int count) {

        log.debug("当前第{}次中转消息, message: {}", count + 1, JsonUtils.toJson(message));

        Boolean lock;
        if (++count > SessionHolder.countRepeaterSession()) {
            // 控制告警间隔
//            lock = redisTemplate.opsForValue().setIfAbsent(CacheConstants.SEND_TO_REPEATER_ERR, "1", Duration.ofMinutes(1));
//            if (!ObjectUtils.isEmpty(lock) && lock) {
//                FeiShuUtil.sendMsg(String.format("【IM Server】 - 连续%s次中转消息失败!!!", count));
//            }
            return;
        }

        // 获取nacos上的健康实例
        ServiceInstance instance = loadBalancerClient.choose(settingConfig.getRepeaterServiceName());

        // 当前没有健康实例 - 发送飞书告警
        if (ObjectUtils.isEmpty(instance)) {
            log.error("中转消息失败, 当前没有可用的repeater实例");
            // 控制告警间隔
//            lock = redisTemplate.opsForValue().setIfAbsent(CacheConstants.SEND_TO_REPEATER_ERR, "1", Duration.ofMinutes(1));
//            if (!ObjectUtils.isEmpty(lock) && lock) {
//                FeiShuUtil.sendMsg("【IM Server】 - 中转消息失败, 当前没有可用的repeater实例!!!");
//            }
            return;
        }

        // 查找该健康实例的session
        String repeaterId = String.format("%s:%s", instance.getHost(), instance.getPort());
        log.debug("中转消息, repeaterId: {}, messageId: {}", repeaterId, message.getMessageId());
        Session session = SessionHolder.getRepeaterSession(repeaterId);

        // 未找到该实例的长连接
        if (ObjectUtils.isEmpty(session)) {
            //递归查找下一个健康实例, 负载均衡的规则是轮询, 所以最大重试次数为与server连接的repeater实例数即可
            sendToRepeater(message, count);
        } else {

            NetkitFuture future = session.sendMessage(message);

            future.addFutureListener(new NetkitFutureListener() {
                @Override
                public void onComplete(boolean success, Session session) {
                    if (success) {
                        log.debug("中转消息成功, message: {}", JsonUtils.toJson(message));
                    } else {
                        log.debug("中转消息失败, repeaterId: {}, message: {}", repeaterId, JsonUtils.toJson(message));
                    }
                }
            });

        }

    }

}
